package com.iteaj.iot.test.mqtt;

import com.iteaj.iot.Message;
import com.iteaj.iot.client.ClientProtocolHandle;
import com.iteaj.iot.client.mqtt.MqttClient;
import com.iteaj.iot.client.mqtt.MqttConnectProperties;
import com.iteaj.iot.consts.ExecStatus;
import com.iteaj.iot.test.IotTestHandle;
import com.iteaj.iot.test.IotTestProperties;
import com.iteaj.iot.test.TestConst;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.time.Duration;
import java.time.Instant;

/**
 * create time: 2021/9/3
 *
 * @author iteaj
 * @since 1.0
 */
public class MqttClientTestHandle implements ClientProtocolHandle<MqttPublishTestProtocol>, IotTestHandle {

    @Autowired
    private IotTestProperties properties;
    @Autowired
    private ThreadPoolTaskScheduler scheduler;
    @Autowired
    private MqttClientTestComponent component;
    private Logger logger = LoggerFactory.getLogger(getClass());
    public static final String IotMqttWillClient = "willClient";

    public static final String TOPIC_RESPONSE = "/test/response";
    public static final String AT_MOST_ONCE_TOPIC = "/test/atMostOnce";
    public static final String EXACTLY_ONCE_TOPIC = "/test/exactlyOnce";
    public static final String AT_LEAST_ONCE_TOPIC = "/test/atLeastOnce";

    @Override
    public Object handle(MqttPublishTestProtocol protocol) {
        return null;
    }

    @Override
    public void start() {
        IotTestProperties.TestMqttConnectProperties config = properties.getMqtt();

        MqttConnectProperties willTopicConfig = new IotTestProperties.TestMqttConnectProperties();
        // 保留遗嘱的测试客户端
        BeanUtils.copyProperties(config, willTopicConfig, "clientId", "deviceSn");
        willTopicConfig.setWillTopic("/willTopic/"+willTopicConfig.getClientId());
        willTopicConfig.setWillRetain(false); // 不保留的遗嘱
        willTopicConfig.setWillMessage("{\"retain\": false}"); // 不保留遗嘱测试
        MqttClient retainWillClient = component.createNewClientAndConnect(willTopicConfig);

        MqttConnectProperties willRetainTopicConfig = new IotTestProperties.TestMqttConnectProperties();
        // 不保留遗嘱的测试客户端
        BeanUtils.copyProperties(config, willRetainTopicConfig, "clientId", "deviceSn");
        willRetainTopicConfig.setWillTopic("/willTopic/"+willRetainTopicConfig.getClientId());
        willRetainTopicConfig.setWillRetain(true); // 保留的遗嘱
        willRetainTopicConfig.setWillMessage("{\"retain\": true}"); // 保留遗嘱测试
        MqttClient willClient = component.createNewClientAndConnect(willRetainTopicConfig);

        scheduler.schedule(() -> {
            System.out.println("---------------------------------------------------- mqtt测试 ----------------------------------------------------------");

            willClient.disconnect(false); // 断线重连测试
            retainWillClient.disconnect(false); // 断线重连测试

            component.clients().forEach(item -> {
                MqttClient mqttClient = (MqttClient) item;

                // 默认客户端用来做订阅测试, 不做发布测试
                if(item == component.getClient()) return;

                IotTestProperties.TestMqttConnectProperties clientConfig = (IotTestProperties.TestMqttConnectProperties) mqttClient.getConfig();
                // 测试最多发送一次报文
                new MqttPublishTestProtocol(MqttQoS.AT_MOST_ONCE, AT_MOST_ONCE_TOPIC, clientConfig.getDeviceSn()).request(clientConfig, protocol -> {
                    final Message.MessageHead head = protocol.requestMessage().getHead();
                    if(protocol.getExecStatus() == ExecStatus.success) {
                        logger.info(TestConst.LOGGER_PROTOCOL_DESC, "mqtt协议", MqttQoS.AT_MOST_ONCE + " -> resp"
                                , head.getEquipCode(), head.getMessageId(), "通过" );
                    } else {
                        logger.error(TestConst.LOGGER_PROTOCOL_DESC, "mqtt协议", MqttQoS.AT_MOST_ONCE + " -> resp"
                                , head.getEquipCode(), head.getMessageId(), protocol.getExecStatus().desc);
                    }
                    return null;
                });

                // 最少发送一次报文
                new MqttPublishTestProtocol(MqttQoS.AT_LEAST_ONCE, AT_LEAST_ONCE_TOPIC, clientConfig.getDeviceSn()).request(clientConfig, protocol -> {
                    final Message.MessageHead head = protocol.requestMessage().getHead();
                    if(protocol.getExecStatus() == ExecStatus.success) {
                        logger.info(TestConst.LOGGER_PROTOCOL_DESC, "mqtt协议", MqttQoS.AT_LEAST_ONCE + " -> resp"
                                , head.getEquipCode(), head.getMessageId(), "通过" );
                    } else {
                        logger.error(TestConst.LOGGER_PROTOCOL_DESC, "mqtt协议", MqttQoS.AT_LEAST_ONCE + " -> resp"
                                , head.getEquipCode(), head.getMessageId(), protocol.getExecStatus().desc);
                    }
                    return null;
                });

                // 确保一定发送一次
                new MqttPublishTestProtocol(MqttQoS.EXACTLY_ONCE, EXACTLY_ONCE_TOPIC, clientConfig.getDeviceSn()).request(clientConfig, protocol -> {
                    final Message.MessageHead head = protocol.requestMessage().getHead();
                    if(protocol.getExecStatus() == ExecStatus.success) {
                        logger.info(TestConst.LOGGER_PROTOCOL_DESC, "mqtt协议", MqttQoS.EXACTLY_ONCE + " -> resp"
                                , head.getEquipCode(), head.getMessageId(), "通过" );
                    } else {
                        logger.error(TestConst.LOGGER_PROTOCOL_DESC, "mqtt协议", MqttQoS.EXACTLY_ONCE + " -> resp"
                                , head.getEquipCode(), head.getMessageId(), protocol.getExecStatus().desc);
                    }
                    return null;
                });

                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }, Instant.now().plusSeconds(10));

        scheduler.scheduleAtFixedRate(() -> {
            try {

                // 测试 client->broker->client->broker->client
//                    new MqttPublishTestProtocol(MqttQoS.AT_MOST_ONCE, "broker").request(protocol -> {
//                        if(protocol.getExecStatus() == ExecStatus.success) {
//                            final Message.MessageHead head = protocol.responseMessage().getHead();
//                            logger.info(TestConst.LOGGER_PROTOCOL_DESC, "mqtt协议", "broker主动请求测试"
//                                    , head.getEquipCode(), head.getMessageId(), "通过");
//                        } else {
//                            final Message.MessageHead head = protocol.requestMessage().getHead();
//                            logger.error(TestConst.LOGGER_PROTOCOL_DESC, "mqtt协议", "broker主动请求测试"
//                                    , head.getEquipCode(), head.getMessageId(), protocol.getExecStatus().desc);
//                        }
//                        return null;
//                    });


            } catch (Exception e) {
                e.printStackTrace();
            }

        }, Instant.now().plusSeconds(18), Duration.ofSeconds(30));
    }

    @Override
    public int getOrder() {
        return 1000 * 30;
    }
}
